package com.demo.function.arithmetic.timewheel;

import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.DelayQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * https://www.webkfz.com/rjjc/m3z9.html
 * Kafka 时间轮算法 java版
 */
@Slf4j
public class SystemTimer implements Timer {
    /**
     * 底层时间轮
     */
    private TimeWheel timeWheel;
    /**
     * 一个Timer只有一个延时队列,传递到每个层级的时间轮中
     */
    private DelayQueue<TimerTaskList> delayQueue = new DelayQueue<>();
    /**
     * 过期任务执行线程
     */
    private ExecutorService workerThreadPool;
    /**
     * 轮询delayQueue获取过期任务线程
     */
    private ExecutorService bossThreadPool;

    public SystemTimer() {
        // 创建时间轮，时间槽单位为5秒，时间槽为12个
        this.timeWheel = new TimeWheel(5000, 12, System.currentTimeMillis(), delayQueue);
        this.workerThreadPool = Executors.newFixedThreadPool(100);
        this.bossThreadPool = Executors.newFixedThreadPool(1);
        // 1000ms推动一次时间轮运转
        this.bossThreadPool.submit(() -> {
            for (; ; ) {
                log.info("当前时间戳：" + System.currentTimeMillis());
                this.advanceClock(1000);
            }
        });
    }

    public void addTimerTaskEntry(TimerTaskEntry entry) {
        if (!timeWheel.add(entry)) {
            //已经过期了
            TimerTask timerTask = entry.getTimerTask();
            log.info("=====任务:{} 已到期,准备执行============",timerTask.getDesc());
            workerThreadPool.submit(timerTask);
        }
    }

    @Override
    public void add(TimerTask timerTask) {
        TimerTaskEntry entry = new TimerTaskEntry(timerTask, timerTask.getDelayMs() + System.currentTimeMillis());
        log.info("=======添加任务开始====task:{}, 延时时间：{}, 超时时间:{}", entry.getTimerTask().getDesc(), entry.getTimerTask().getDelayMs(), entry.getExpireMs());
        timerTask.setTimerTaskEntry(entry);
        addTimerTaskEntry(entry);
    }

    /**
     * 推动指针运转获取过期任务
     *
     * @param timeout 时间间隔
     * @return
     */
    @Override
    public synchronized void advanceClock(long timeout) {
        try {
            TimerTaskList bucket = delayQueue.poll(timeout, TimeUnit.MILLISECONDS);
            log.info("bucket：" + JSON.toJSONString(bucket));
            if (bucket != null) {
                //推进时间
                timeWheel.advanceLock(bucket.getExpiration());
                //把槽中所有任务都取出来，并重新add一遍，此时因为任务到期，并不会真正add进去，而是调用线程池运行任务，具体看下面的reinsert和TimingWheel.add方法
                bucket.clear(this::addTimerTaskEntry);
            }
        } catch (InterruptedException e) {
            log.error("advanceClock error");
        }
    }

    @Override
    public int size() {
        //todo
        return 0;
    }

    @Override
    public void shutdown() {
        this.bossThreadPool.shutdown();
        this.workerThreadPool.shutdown();
        this.timeWheel = null;
    }
}
